package com.shujia.streaming

import com.mysql.jdbc.Driver
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{Connection, DriverManager, ResultSet, Statement}
import scala.collection.mutable.ListBuffer

object Code06Watcher {
  def main(args: Array[String]): Unit = {

    // 如果要使用SparkStreaming 需要获取其操作对象 => StreamingContext
    // 如果创建StreamingContext对象那么需要SparkContext对象
    // local[2] 表示本地模式 并且设置并发数为2 此项为必须项
    val sc = new SparkContext(new SparkConf().setMaster("local[2]").setAppName("wordCountStreaming"))

    // Durations.seconds(5) 表示每5秒启动一个计算任务
    val streamingContext: StreamingContext = new StreamingContext(sc, Durations.seconds(5))
    // 可以监听socket端口获取数据 master:8888 在Linux中可以通过nc -lk master 8888 启动
    val stream: ReceiverInputDStream[String] = streamingContext.socketTextStream("master", 8888)


    // 方式1：程序运行时，是需要在生产环境下运行，但是该List方式是在代码中写定  => 改进： 将数据保存到Mysql中，再从Mysql中获取
    //    val list: List[String] = List("47BE1E866CFC071DB19D5E1C056BE28AE24C16E7", "D55433A437AEC8D8D3DB2BCA56E9E64392A9D93C", "27A43BC5BB6853DE779A9EA7F3523733CACBD21C")

    // 方式2：该方式在应用程序刚启动时，会执行一次，之后对数据进行添加，不会生效
    //    val list: ListBuffer[String] = ListBuffer()
    //        DriverManager.registerDriver(new Driver())
    //        val connection: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student?useSSL=false", "root", "123456")
    //        val statement: Statement = connection.createStatement()
    //
    //        val resultSet: ResultSet = statement.executeQuery(
    //          """
    //            |SELECT * FROM watcher
    //            |""".stripMargin)
    //
    //        while (resultSet.next()){
    //          val mdn: String = resultSet.getString("mdn")
    //          list.append(mdn)
    //        }
    //
    //
    //    stream
    //      .map((line: String) => {
    //        val splitRes: Array[String] = line.split(",")
    //        // 实时数据：MDN,网格ID,省编号，区县编号，网格停留时间，开始时间，结束时间等
    //        (splitRes(0), splitRes(1), splitRes(2), splitRes(3), splitRes(4), splitRes(5), splitRes(6))
    //      })
    //      .filter {
    //        case (mdn, grid_id, city_id, county_id, duration, start_time, end_time) => {
    //          list.contains(mdn)
    //        }
    //      }
    //      .print()


    // 方式3：对于如下方式，当数据量较多时，会导致创建过多Mysql链接 每条数据都会连接一次Mysql
    //    stream
    //      .map((line: String) => {
    //        val splitRes: Array[String] = line.split(",")
    //        // 实时数据：MDN,网格ID,省编号，区县编号，网格停留时间，开始时间，结束时间等
    //        (splitRes(0), splitRes(1), splitRes(2), splitRes(3), splitRes(4), splitRes(5), splitRes(6))
    //      })
    //      .filter {
    //        case (mdn, grid_id, city_id, county_id, duration, start_time, end_time) => {
    //          DriverManager.registerDriver(new Driver())
    //          val connection: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student?useSSL=false", "root", "123456")
    //          val statement: Statement = connection.createStatement()
    //
    //          val resultSet: ResultSet = statement.executeQuery(
    //            s"""
    //              |SELECT * FROM watcher where mdn = '$mdn'
    //              |""".stripMargin)
    //
    //          var flag:Boolean = false
    //          while (resultSet.next()) {
    //            flag = true
    //          }
    //          flag
    //        }
    //      }
    //      .print()


    // 方式4：对于每个批次执行一次 Mysql查询
    // transform是对于每个批次数据形成一个RDD并执行一次
    stream.transform(
      // 就相当于 一个SparkCore的Driver
      (rdd: RDD[String]) => {
        // 先创建Mysql链接
        val list: ListBuffer[String] = ListBuffer()
        DriverManager.registerDriver(new Driver())
        val connection: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student?useSSL=false", "root", "123456")
        val statement: Statement = connection.createStatement()

        val resultSet: ResultSet = statement.executeQuery(
          s"""
             |SELECT * FROM watcher
             |""".stripMargin)
        while (resultSet.next()) {
          val mdn: String = resultSet.getString("mdn")
          list.append(mdn)
        }
        // list保存有数据
        // 使用广播变量对计算过程进行优化
        val broadcastList: Broadcast[ListBuffer[String]] = sc.broadcast(list)

        rdd
          .map((line: String) => {
            val splitRes: Array[String] = line.split(",")
            // 实时数据：MDN,网格ID,省编号，区县编号，网格停留时间，开始时间，结束时间等
            (splitRes(0), splitRes(1), splitRes(2), splitRes(3), splitRes(4), splitRes(5), splitRes(6))
          })
          .filter {
            case (mdn, grid_id, city_id, county_id, duration, start_time, end_time) => {
              broadcastList.value.contains(mdn)
            }
          }
      }
    ).print()


    // 需要启动 streamingContext
    streamingContext.start()
    // 等待停止信号 <- 手动停止
    streamingContext.awaitTermination()

    // Streaming程序是 7 * 24 小时不间断运行，那么不需停止
    //    streamingContext.stop()

  }
}
